package com.shujia.base;

import com.shujia.bean.ComparableName;
import com.shujia.bean.FilterName;
import com.shujia.utils.HBaseUtil;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.CompareOperator;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.io.BufferedReader;
import java.io.FileReader;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.Properties;

public class HBaseFilter {
    private Connection conn;
    private Admin admin;

    /**
     * 在所有Test方法执行之前执行
     */
    @Before
    public void getConnection() {
        conn = HBaseUtil.CONNECTION;
        admin = HBaseUtil.ADMIN;
    }


    /**
     * 需求：通过RowFilter与BinaryComparator过滤比rowKey 1500100010小的所有值出来
     * 配合等值比较器BinaryComparator
     */
    @Test
    public void scanDataWithRowFilterAndBinaryComparator() {
        try {
            /**
             * CompareFilter 是一个抽象类，无法被实例化
             * 但是，我们通过观察源码，可以得到一些信息，创建比较过滤器对象的时候，至少需要传入两种参数：操作符，比较器
             */
            //获取表对象
            TableName st = TableName.valueOf("students");
            Table students = conn.getTable(st);
            Scan scan = new Scan();

            //创建一个BinaryComparator比较器对象
            BinaryComparator binaryComparator = new BinaryComparator(Bytes.toBytes("1500100010"));

            //创建一个行键过滤器对象
            //旧版本创建方式：public RowFilter(final CompareOp rowCompareOp,final ByteArrayComparable rowComparator)
//            RowFilter rowFilter = new RowFilter(CompareFilter.CompareOp.LESS, binaryComparator);
            //新版本创建方式：public RowFilter(final CompareOperator op,final ByteArrayComparable rowComparator)
            RowFilter rowFilter = new RowFilter(CompareOperator.LESS, binaryComparator);

            //设置过滤器
            //public Scan setFilter(Filter filter)
            scan.setFilter(rowFilter);

            ResultScanner resultScanner = students.getScanner(scan);
            for (Result result : resultScanner) {
                HBaseUtil.printResult(result);
            }

        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 列簇过滤器：FamilyFilter
     * 需求：过滤出列簇名中包含“a”的列簇中所有数据配合包含比较器SubstringComparator
     */
    @Test
    public void scanDataWithFamilyFilterAndSubstringComparator() {
        try {
            /**
             * CompareFilter 是一个抽象类，无法被实例化
             * 但是，我们通过观察源码，可以得到一些信息，创建比较过滤器对象的时候，至少需要传入两种参数：操作符，比较器
             */
            //获取表对象
            TableName st = TableName.valueOf("users");
            Table students = conn.getTable(st);
            Scan scan = new Scan();

            //创建一个包含比较器
            SubstringComparator substringComparator = new SubstringComparator("f");

            //创建列簇过滤器
            //public FamilyFilter(final CompareOperator op, final ByteArrayComparable familyComparator)
            FamilyFilter familyFilter = new FamilyFilter(CompareOperator.EQUAL, substringComparator);


            //设置过滤器
            //public Scan setFilter(Filter filter)
            scan.setFilter(familyFilter);

            ResultScanner resultScanner = students.getScanner(scan);
            for (Result result : resultScanner) {
                HBaseUtil.printResult(result);
            }

        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 列名过滤器：QualifierFilter
     * 通过QualifierFilter与SubstringComparator查询列名包含 m 的列的值
     */
    @Test
    public void scanDataWithQualifierFilterAndSubstringComparator() {
        try {
            /**
             * CompareFilter 是一个抽象类，无法被实例化
             * 但是，我们通过观察源码，可以得到一些信息，创建比较过滤器对象的时候，至少需要传入两种参数：操作符，比较器
             */
            //获取表对象
            TableName st = TableName.valueOf("users");
            Table students = conn.getTable(st);
            Scan scan = new Scan();

            //创建一个包含比较器
            SubstringComparator substringComparator = new SubstringComparator("m");

            //创建列名过滤器
            //public QualifierFilter(final CompareOperator op, final ByteArrayComparable qualifierComparator)
            QualifierFilter qualifierFilter = new QualifierFilter(CompareOperator.EQUAL, substringComparator);

            //设置过滤器
            //public Scan setFilter(Filter filter)
            scan.setFilter(qualifierFilter);

            ResultScanner resultScanner = students.getScanner(scan);
            for (Result result : resultScanner) {
                HBaseUtil.printResult(result);
            }

        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 列值过滤器(属于比较过滤器)
     * 通过ValueFilter与BinaryPrefixComparator过滤出所有的cell中值以 "从" 开头的学生
     * 注意：列值过滤器只能过滤出当前符合条件的列，至于与该列属于同一行的其他列并不会返回
     */
    @Test
    public void scanDataWithValueFilterAndBinaryPrefixComparator() {
        try {
            /**
             * CompareFilter 是一个抽象类，无法被实例化
             * 但是，我们通过观察源码，可以得到一些信息，创建比较过滤器对象的时候，至少需要传入两种参数：操作符，比较器
             */
            //获取表对象
            TableName st = TableName.valueOf("students");
            Table students = conn.getTable(st);
            Scan scan = new Scan();

            //创建二进制前缀比较器
            BinaryPrefixComparator binaryPrefixComparator = new BinaryPrefixComparator(Bytes.toBytes("从"));

            //创建列值过滤器对象
            //public ValueFilter(final CompareOperator valueCompareOp, final ByteArrayComparable valueComparator)
            ValueFilter valueFilter = new ValueFilter(CompareOperator.EQUAL, binaryPrefixComparator);

            //设置过滤器
            //public Scan setFilter(Filter filter)
            scan.setFilter(valueFilter);

            ResultScanner resultScanner = students.getScanner(scan);
            for (Result result : resultScanner) {
                HBaseUtil.printResult(result);
            }

        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 单列值过滤器(专用过滤器)
     * 需求：通过SingleColumnValueFilter与BinaryPrefixComparator过滤出所有的cell中值以 "从" 开头的学生
     */
    @Test
    public void scanDataWithSingleColumnValueFilterAndBinaryPrefixComparator() {
        try {
            /**
             * CompareFilter 是一个抽象类，无法被实例化
             * 但是，我们通过观察源码，可以得到一些信息，创建比较过滤器对象的时候，至少需要传入两种参数：操作符，比较器
             */
            //获取表对象
            TableName st = TableName.valueOf("students");
            Table students = conn.getTable(st);
            Scan scan = new Scan();

            //创建二进制前缀比较器
            BinaryPrefixComparator binaryPrefixComparator = new BinaryPrefixComparator(Bytes.toBytes("从"));

            //创建单列值过滤器
            //public SingleColumnValueFilter(final byte [] family, final byte [] qualifier,
            //      final CompareOperator op,
            //      final org.apache.hadoop.hbase.filter.ByteArrayComparable comparator)
            SingleColumnValueFilter singleColumnValueFilter = new SingleColumnValueFilter(Bytes.toBytes("info"), Bytes.toBytes("name"),
                    CompareOperator.EQUAL, binaryPrefixComparator);

            //设置过滤器
            //public Scan setFilter(Filter filter)
            scan.setFilter(singleColumnValueFilter);

            ResultScanner resultScanner = students.getScanner(scan);
            for (Result result : resultScanner) {
                HBaseUtil.printResult(result);
            }

        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 列值排除过滤器
     * SingleColumnValueExcludeFilter
     * 通过SingleColumnValueExcludeFilter与BinaryComparator查询文科一班所有学生信息，最终不返回clazz列
     */
    @Test
    public void scanDataWithSingleColumnValueExcludeFilterAndBinaryComparator() {
        try {
            /**
             * CompareFilter 是一个抽象类，无法被实例化
             * 但是，我们通过观察源码，可以得到一些信息，创建比较过滤器对象的时候，至少需要传入两种参数：操作符，比较器
             */
            //获取表对象
            TableName st = TableName.valueOf("students");
            Table students = conn.getTable(st);
            Scan scan = new Scan();

            //创建二进制内容比较器
            BinaryComparator binaryComparator = new BinaryComparator(Bytes.toBytes("文科一班"));

            //创建列值排除过滤器
            //public SingleColumnValueExcludeFilter(byte[] family, byte[] qualifier,
            //                                        CompareOperator op, ByteArrayComparable comparator)
            SingleColumnValueExcludeFilter singleColumnValueExcludeFilter = new SingleColumnValueExcludeFilter(Bytes.toBytes("info"), Bytes.toBytes("clazz"),
                    CompareOperator.EQUAL, binaryComparator);

            //设置过滤器
            //public Scan setFilter(Filter filter)
            scan.setFilter(singleColumnValueExcludeFilter);

            ResultScanner resultScanner = students.getScanner(scan);
            for (Result result : resultScanner) {
                HBaseUtil.printResult(result);
            }

        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 行键前缀过滤器  PrefixFilter
     * 通过PrefixFilter查询以15001001开头的所有前缀的rowkey
     */
    @Test
    public void scanDataWithPrefixFilter() {
        try {
            /**
             * CompareFilter 是一个抽象类，无法被实例化
             * 但是，我们通过观察源码，可以得到一些信息，创建比较过滤器对象的时候，至少需要传入两种参数：操作符，比较器
             */
            //获取表对象
            TableName st = TableName.valueOf("students");
            Table students = conn.getTable(st);
            Scan scan = new Scan();

            //PrefixFilter(final byte [] prefix)
            PrefixFilter prefixFilter = new PrefixFilter(Bytes.toBytes("150010093"));

            //设置过滤器
            //public Scan setFilter(Filter filter)
            scan.setFilter(prefixFilter);

            ResultScanner resultScanner = students.getScanner(scan);
            for (Result result : resultScanner) {
                HBaseUtil.printResult(result);
            }

        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 分页过滤器
     * PageFilter
     * 通过PageFilter查询三页的数据，每页10条
     */
    @Test
    public void scanDataWithPageFilter() {
        try {
            /**
             * CompareFilter 是一个抽象类，无法被实例化
             * 但是，我们通过观察源码，可以得到一些信息，创建比较过滤器对象的时候，至少需要传入两种参数：操作符，比较器
             */
            //获取表对象
            TableName st = TableName.valueOf("students");
            Table students = conn.getTable(st);

            //设置几个变量
            //设置要查询的页数
            int pageNum = 5;
            //设置每一页要查询的条数
            int pageNumData = 10;
            //设置开始行
            String startRowKey = "";

            //创建分页过滤器对象
            PageFilter pageFilter = new PageFilter(pageNumData);
            //创建Scan的对象
            Scan scan = new Scan();
            scan.setFilter(pageFilter);

            for (int i = 1; i <= pageNum; i++) {
                System.out.println("====================当前是第 " + i + " 页=========================");
                //查询10条数据
                ResultScanner resultScanner = students.getScanner(scan);
                //获取该页的最后一行的行键
                for (Result result : resultScanner) {
                    startRowKey = Bytes.toString(result.getRow()) + 0;
                    //设置下一行的开始
                    scan.withStartRow(Bytes.toBytes(startRowKey));
                    HBaseUtil.printResult(result);
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 包装过滤器之SkipFilter过滤器
     *
     * 1500100071,祖昌勋,22,男,理科四班
     */
    @Test
    public void scanDataWithSkipFilter(){
        try {
            /**
             * CompareFilter 是一个抽象类，无法被实例化
             * 但是，我们通过观察源码，可以得到一些信息，创建比较过滤器对象的时候，至少需要传入两种参数：操作符，比较器
             */
            //获取表对象
            TableName st = TableName.valueOf("students");
            Table students = conn.getTable(st);
            Scan scan = new Scan();

            ValueFilter valueFilter = new ValueFilter(CompareOperator.NOT_EQUAL, new SubstringComparator("理科"));

            //创建跳过过滤器
            //SkipFilter(Filter filter)
            SkipFilter skipFilter = new SkipFilter(valueFilter);

            scan.setFilter(skipFilter);

            ResultScanner resultScanner = students.getScanner(scan);
            for (Result result : resultScanner) {
                HBaseUtil.printResult(result);
            }

        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 包装过滤器之WhileMatchFilter过滤器
     *
     * 1500100071,祖昌勋,22,男,理科四班
     */
    @Test
    public void scanDataWithWhileMatchFilter(){
        try {
            /**
             * CompareFilter 是一个抽象类，无法被实例化
             * 但是，我们通过观察源码，可以得到一些信息，创建比较过滤器对象的时候，至少需要传入两种参数：操作符，比较器
             */
            //获取表对象
            TableName st = TableName.valueOf("students");
            Table students = conn.getTable(st);
            Scan scan = new Scan();

            ValueFilter valueFilter = new ValueFilter(CompareOperator.NOT_EQUAL, new SubstringComparator("祖昌勋"));

            //创建跳过过滤器
            //SkipFilter(Filter filter)
            WhileMatchFilter whileMatchFilter = new WhileMatchFilter(valueFilter);

            scan.setFilter(whileMatchFilter);

            ResultScanner resultScanner = students.getScanner(scan);
            for (Result result : resultScanner) {
                HBaseUtil.printResult(result);
            }

        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 组合过滤器
     * 需求：通过运用4种比较器过滤出姓于，年纪大于22岁，性别为女，且是理科的学生。
     */
    @Test
    public void scanDataWithGroupFilter(){
        try {
            /**
             * CompareFilter 是一个抽象类，无法被实例化
             * 但是，我们通过观察源码，可以得到一些信息，创建比较过滤器对象的时候，至少需要传入两种参数：操作符，比较器
             */
            //获取表对象
            TableName st = TableName.valueOf("students");
            Table students = conn.getTable(st);
            Scan scan = new Scan();

            //创建单列值过滤器
            //过滤出姓于的
            SingleColumnValueFilter singleColumnValueFilter = new SingleColumnValueFilter(Bytes.toBytes("info"), Bytes.toBytes("name"),
                    CompareOperator.EQUAL, new BinaryPrefixComparator(Bytes.toBytes("于")));

            //年纪大于22岁
            SingleColumnValueFilter singleColumnValueFilter1 = new SingleColumnValueFilter(Bytes.toBytes("info"), Bytes.toBytes("age"),
                    CompareOperator.GREATER, new BinaryComparator(Bytes.toBytes("22")));

            //性别为女
            SingleColumnValueFilter singleColumnValueFilter2 = new SingleColumnValueFilter(Bytes.toBytes("info"), Bytes.toBytes("gender"),
                    CompareOperator.EQUAL, new BinaryComparator(Bytes.toBytes("女")));

            //理科
            SingleColumnValueFilter singleColumnValueFilter3 = new SingleColumnValueFilter(Bytes.toBytes("info"), Bytes.toBytes("clazz"),
                    CompareOperator.EQUAL, new SubstringComparator("理科"));

            //如何以每次添加一个过滤器作为条件的话，后一次添加的过滤器逻辑会将前一次覆盖
//            scan.setFilter(singleColumnValueFilter);
//            scan.setFilter(singleColumnValueFilter1);
//            scan.setFilter(singleColumnValueFilter2);
//            scan.setFilter(singleColumnValueFilter3);

            //通过观察源码发现，虽然setFilter这个方法只能传Filter的子类对象
            //但是我们发现了该抽象类的一个子类FilterList表示多个过滤器集合的意思
            //所以我们可以先创建FilterList对象，将多个过滤器封装成List集合添加到FilterList对象中，最终将FilterList对象再添加到scan中
            //先将上面4个过滤器添加到List集合中
            ArrayList<Filter> filters = new ArrayList<>();
            filters.add(singleColumnValueFilter);
            filters.add(singleColumnValueFilter1);
            filters.add(singleColumnValueFilter2);
            filters.add(singleColumnValueFilter3);

            //创建FilterList对象
            FilterList filterList = new FilterList();
            //public void addFilter(List<Filter> filters)
            filterList.addFilter(filters);


            scan.setFilter(filterList);


            ResultScanner resultScanner = students.getScanner(scan);
            for (Result result : resultScanner) {
                HBaseUtil.printResult(result);
            }

        } catch (Exception e) {
            e.printStackTrace();
        }
    }


    @Test
    public void testHBaseUtil() {
        //需求：通过RowFilter与BinaryComparator过滤比rowKey 1500100010小的所有值出来
//        HBaseUtil.scanDataWithCompareFilter("students", FilterName.ROWKEY,
//                CompareOperator.LESS,ComparableName.BINARY,"1500100010");

        //需求：过滤出列簇名中包含“a”的列簇中所有数据配合包含比较器SubstringComparator
//        HBaseUtil.scanDataWithCompareFilter("users",FilterName.FAMILY,
//                CompareOperator.EQUAL,ComparableName.SUBSTRING,"a");

        //需求：通过QualifierFilter与SubstringComparator查询列名包含 m 的列的值
//        HBaseUtil.scanDataWithCompareFilter("users",FilterName.QUALIFIER,
//                CompareOperator.EQUAL,ComparableName.SUBSTRING,"m");

        //需求：通过ValueFilter与BinaryPrefixComparator过滤出所有的cell中值以 "从" 开头的学生
        HBaseUtil.scanDataWithCompareFilter("students", FilterName.VALUE,
                CompareOperator.EQUAL, ComparableName.BINARY_PREFIX, "从");

    }


    /**
     * 释放Before创建的资源，在每个Test之后执行
     */
    @After
    public void closeSource() {
        HBaseUtil.closeSource();
    }
}
